package com.northpool.operator.analysis.dataset;

import com.northpool.operator.annotation.InputParam;
import com.northpool.operator.annotation.InputParams;
import com.northpool.operator.annotation.OperatorInfo;
import com.northpool.operator.annotation.ParamEnum;
import com.northpool.operator.annotation.ParamEnums;
import com.northpool.operator.common.stpclient.StpClientOperatorBase;
import com.northpool.operator.param.ParamTool;
import com.northpool.operator.param.custom.DatasetParam;
import com.northpool.operator.type.TypeEnum;
import com.northpool.stpclient.bean.BaseResponse;
import com.northpool.stpclient.modules.geoprocess.operator.IPipelineOperator;
import com.northpool.stpclient.operator.bean.DatasetBean;
import com.northpool.stpclient.operator.bean.OperatorBean;
import com.northpool.stpclient.operator.bean.ParamBean;
import com.northpool.stpclient.operator.bean.PipelineBean;

import java.io.File;
import java.util.LinkedList;
import java.util.List;

@OperatorInfo(
        ref = "SpaceConnOperator",
        name = "空间连接",
        description = "空间连接",
        enums = {
                @ParamEnums(ref = "matchOption", value = {
                        @ParamEnum(ref = "INTERSECT", name = "相交", description = "如果连接要素与目标要素相交，将匹配连接要素中相交的要素"),
                        @ParamEnum(ref = "CONTAINS", name = "包含", description = "如果目标要素中包含连接要素中的要素，将匹配连接要素中被包含的要素。 目标要素必须是面或折线"),
                        @ParamEnum(ref = "WITHIN", name = "内含", description = "如果目标要素位于连接要素内，将匹配连接要素中包含目标要素的要素。 这与 CONTAINS 相反"),
                }),
                @ParamEnums(ref = "joinOperation", value = {
                        @ParamEnum(ref = "ONE_TO_ONE", name = "关联后取第一个值", description = "关联后取第一个值"),
                        @ParamEnum(ref = "ONE_TO_MANY", name = "关联后拆分为多行数据", description = "关联后拆分为多行数据"),
                }),
        }
)
public class SpaceConnOperator extends StpClientOperatorBase {
    String dataSourceIdA;
    String tableNameA;
    String spatialFieldA;
    String sqlFilterA;
    String dataSourceIdB;
    String tableNameB;
    String spatialFieldB;
    String sqlFilterB;
    String targetDataSourceId;
    String targetTableName;
    String matchOption;
    String fieldMapping;
    String joinFields;
    String joinOperation;

    @InputParams({
            @InputParam(ref = "datasetA", name = "数据集A", desc = "数据集包含数据源ID和数据表名称的JSON描述，如: {\"dataSourceId\":\"xxx-xxx\",\"tableName\":\"table\"}", type = TypeEnum.DB, required = true),
            @InputParam(ref = "sqlFilterA", name = "SQL过滤条件A", desc = "SQL过滤条件, 如: dlmc = '旱地'", type = TypeEnum.STRING, required = false, testValue = ""),
            @InputParam(ref = "datasetB", name = "数据集B", desc = "数据集包含数据源ID和数据表名称的JSON描述，如: {\"dataSourceId\":\"xxx-xxx\",\"tableName\":\"table\"}", type = TypeEnum.DB, required = true),
            @InputParam(ref = "joinFields", name = "连接字段", desc = "多个字段逗号隔开", type = TypeEnum.STRING, required = true),
            @InputParam(ref = "sqlFilterB", name = "SQL过滤条件B", desc = "SQL过滤条件, 如: dlmc = '旱地'", type = TypeEnum.STRING, required = false),
            @InputParam(ref = "targetDataSourceId", name = "目标数据源ID", desc = "目标数据源ID", type = TypeEnum.DATASOURCE, required = true),
            @InputParam(ref = "targetTableName", name = "目标表名称", desc = "目标表名称", type = TypeEnum.STRING, required = true),
            @InputParam(ref = "matchOption", name = "匹配条件", type = TypeEnum.STRING, value = "INTERSECT", desc = "指定将用于匹配行的条件，INTERSECT—如果连接要素与目标要素相交，将匹配连接要素中相交的要素;" +
                    "CONTAINS—如果目标要素中包含连接要素中的要素，将匹配连接要素中被包含的要素。 目标要素必须是面或折线;" +
                    "WITHIN—如果目标要素位于连接要素内，将匹配连接要素中包含目标要素的要素。 这与 CONTAINS 相反。"),
            @InputParam(ref = "fieldMapping", name = "字段映射JSON", value = "", desc = "新增字段和原始字段的映射JSON, 例如：{\"dlmc\":\"DLMC\"}", type = TypeEnum.STRING),
            @InputParam(ref = "joinOperation", name = "连接操作类型", value = "ONE_TO_ONE", desc = "枚举值: ONE_TO_ONE:关联后取第一个值;ONE_TO_MANY:关联后拆分为多行数据;", type = TypeEnum.ENUM, enumRef = "joinOperation"),
    })
    public SpaceConnOperator() {
        super();
    }

    @Override
    public void execute(ParamTool params) throws Exception {
        DatasetParam datasetA= params.getDataset("datasetA");
        DatasetParam datasetB = params.getDataset("datasetB");

        this.dataSourceIdA = datasetA.getDataSourceId();
        this.tableNameA = datasetA.getTableName();
        this.sqlFilterA = params.getStr("sqlFilterA");
        this.dataSourceIdB = datasetB.getDataSourceId();
        this.tableNameB = datasetB.getTableName();
        this.sqlFilterB = params.getStr("sqlFilterB");
        this.joinFields = params.getStr("joinFields");
        this.targetDataSourceId = params.getStr("targetDataSourceId");
        this.targetTableName = params.getStr("targetTableName");
        this.matchOption = params.getStr("matchOption");
        this.fieldMapping = params.getStr("fieldMapping");
        this.joinOperation = params.getStr("joinOperation");

        PipelineBean pipeline = new PipelineBean();
        pipeline.setResultFile(System.getProperty("user.dir") + File.separator + "result_file");
        DatasetBean dataSet = new DatasetBean();
        dataSet.setDataSourceId(dataSourceIdA);
        dataSet.setTableName(tableNameA);
        dataSet.setSqlFilter(sqlFilterA);
        pipeline.setDataSet(dataSet);

        List<OperatorBean> ops = new LinkedList<>();
        ops.add(buildMainOperator());
        ops.add(buildPgSink());

        pipeline.setOperators(ops);
        IPipelineOperator iPipelineOperator = this.client.pipelineOperator();
        BaseResponse response = iPipelineOperator.startPipeline(pipeline);
        iPipelineOperator.onStart(response.getRequestId(), () -> {
            this.log("空间连接开始");
        });
        iPipelineOperator.onSuccess(response.getRequestId(), () -> {
            this.log("空间连接成功");
            String log = iPipelineOperator.log(response.getRequestId());
            this.log(log);
            this.successExit();
        });
        iPipelineOperator.onError(response.getRequestId(), () -> {
            this.error("空间连接失败");
            String log = iPipelineOperator.log(response.getRequestId());
            this.error(log);
            this.failedExit();
        });
        this.waiting(24);
    }

    public OperatorBean buildMainOperator() {
        OperatorBean operatorBean = new OperatorBean();
        operatorBean.setClassName("com.northpool.stpprocess.operator.spatial.vector.SpaceConnOperator");

        List<ParamBean> params = new LinkedList<>();
        params.add(new ParamBean("originSpatialField", spatialFieldA));
        params.add(new ParamBean("dataSourceId", dataSourceIdB));
        params.add(new ParamBean("tableName", tableNameB));
        params.add(new ParamBean("spatialField", spatialFieldB));
        params.add(new ParamBean("fields", joinFields));
        params.add(new ParamBean("matchOption", matchOption));
        params.add(new ParamBean("fieldMapping", fieldMapping));
        params.add(new ParamBean("joinOperation", joinOperation));
        operatorBean.setStartParam(params);

        return operatorBean;
    }

    public OperatorBean buildPgSink() {
        OperatorBean operatorBean = new OperatorBean();
        operatorBean.setClassName("com.northpool.stpprocess.operator.basic.sink.PostgresSink");

        List<ParamBean> params = new LinkedList<>();
        params.add(new ParamBean("dataSourceId", targetDataSourceId));
        params.add(new ParamBean("tableName", targetTableName));
        params.add(new ParamBean("templateDataSourceId", dataSourceIdA));
        params.add(new ParamBean("templateTableName", tableNameA));
        params.add(new ParamBean("createPK", "false"));
        params.add(new ParamBean("completionField", "true"));
        operatorBean.setStartParam(params);

        return operatorBean;
    }
}
